package jsen.tool.ddd.adapter.event.demo;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @author jsen
 * @date 2021/10/6
 */
@SuppressWarnings("all")
public class SyncTaskBack<T> {

    private final ExecutorService executorService;
    private final List<CountDownLatch> countDownLatchList = Lists.newArrayList();
    private boolean finished = false;
    private final AtomicInteger total = new AtomicInteger(0);
    private final AtomicInteger completed = new AtomicInteger(0);
    private final Stopwatch stopwatch;
    private RuntimeException exception = null;


    public SyncTaskBack(ExecutorService executorService) {
        this.executorService = executorService;
        this.stopwatch = Stopwatch.createStarted();
    }

    public synchronized void addTask(List<T> taskList, Consumer<T> handler) {
        if (taskList == null || taskList.isEmpty()) {
            return;
        }
        int size = taskList.size();
        total.getAndAdd(size);
        CountDownLatch countDownLatch = new CountDownLatch(size);
        countDownLatchList.add(countDownLatch);
        taskList.forEach(task ->
                        executorService.execute(() -> {
                            try {
                                handler.accept(task);
                            } catch (Exception e) {
//                        log.error("SyncTask handle error", e);
                            } finally {
                                countDownLatch.countDown();
                                completed.getAndAdd(1);
                                progress();
                            }
                        })
        );
    }

    public synchronized void addTaskNoCatch(List<T> taskList, Consumer<T> handler) {
        if (taskList == null || taskList.isEmpty()) {
            return;
        }
        int size = taskList.size();
        total.getAndAdd(size);
        CountDownLatch countDownLatch = new CountDownLatch(size);
        countDownLatchList.add(countDownLatch);
        taskList.forEach(task ->
                        executorService.execute(() -> {
                            try {
                                if (exception != null) {
                                    return;
                                }
                                handler.accept(task);
                            } catch (RuntimeException e) {
//                        log.error("SyncTask handle error", e);
                                exception = e;
                            } finally {
                                countDownLatch.countDown();
                                completed.getAndAdd(1);
                                progress();
                            }
                        })
        );
    }

    public synchronized void sync() {
        if (finished) {
            throw new RuntimeException("SyncTask finished");
        }
        finished = true;
        for (CountDownLatch countDownLatch : countDownLatchList) {
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
//                log.error("InterruptedException", e);
                throw new RuntimeException("CountDownLatch.await()", e);
            }
        }
        if (exception != null) {
//            log.info("任务跑完，结果异常，耗时" + stopwatch.elapsed(TimeUnit.MILLISECONDS));
            throw exception;
        }
//        log.info("任务跑完，结果正常，耗时" + stopwatch.elapsed(TimeUnit.MILLISECONDS));
    }

    private void progress() {
        if (total.get() > 0) {
//            log.info("SyncTask progress {}", (float) completed.get() / (float) total.get());
        }
    }

}
